package com.chb.userbehavioranalysis.traffic

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.apache.hadoop.util.bloom.BloomFilter
import redis.clients.jedis.Jedis

/**
 * Unique User View
 */
object UvWithBloom {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        env.setParallelism(1)


        // 用相对路径定义数据源
        val resource = getClass.getResource("/UserBehavior.csv")
        val dataStream = env.readTextFile(resource.getPath)
            .map(data => {
                val dataArray = data.split(",")
                UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim, dataArray(4).trim.toLong + (2.56 * 3600 * 24 * 365).toLong)
            })
            .assignAscendingTimestamps(_.timestamp * 1000L)
            .filter(_.behavior == "pv") // 只统计pv操作
            .map(data => ("Key", data.userId))
            .keyBy(_._1)
            .timeWindow(Time.hours(1)) // 滚动窗口
            .trigger(new MyTrigger())
            .process(new UvCountWithBoom())
        dataStream.print()

        env.execute("uv with bloom filter job")
    }
}


class UvCountWithBoom() extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow] {
    // 定义redisli连接
    lazy val jedis = new Jedis("10.0.0.201", 56379)
    lazy val boomfilter = new MyBoomFilter(1 << 29)

    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {
        // 位图存储方式 key是windowEnd, value 是bitmap
        val storeKey = context.window.getEnd.toString
        var count = 0L // userId记录数
        if (jedis.hget("count", storeKey) != null) {
            count = jedis.hget("count", storeKey).toLong
        }

        // 使用boomfilter判断userId是否存在
        val userId = elements.iterator.next().toString()
        val offset = boomfilter.hash(userId, 61) // hash

        // 定义一个标志位 判断redis位图中有没有这位
        val isExist = jedis.getbit(storeKey, offset)
        if (!isExist) {
            // 如果不存在，位图对应位置 置成1  count+1
            jedis.hset("count", storeKey, (count + 1).toString)
            jedis.setbit(storeKey, offset, true) //  位图对应位置 置成1
            // 控制台打印
            out.collect(UvCount(storeKey.toLong, count + 1))
        } else {
            // 存在
            out.collect(UvCount(storeKey.toLong, count))
        }
    }
}

class MyTrigger extends Trigger[(String, Long), TimeWindow] {
    val c = 0
    // 定义了触发时间，来一条触发一条
    override def onElement(t: (String, Long), l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {
        // 每来一条数据，就触发窗口操作， 并且清空窗口的状态
        if (c%10000 == 0) {
            TriggerResult.FIRE_AND_PURGE
        } else {
            TriggerResult.CONTINUE
        }

    }

    override def onProcessingTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

    override def onEventTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

    override def clear(w: TimeWindow, triggerContext: Trigger.TriggerContext): Unit = {}
}


// 定义一个布隆过滤器
class MyBoomFilter(size: Long) extends Serializable {
    // 指定位图的大小， 默认16M
    private val cap = if (size > 0) size else 1 << 27

    def hash(value: String, seed: Int): Long = {
        var result = 0L
        for (i <- 0 until value.length) {
            result = result * seed + value.charAt(i)
        }
        result & (cap - 1)
    }
}
